package com.link.serve.kafka.factory;

import java.util.HashMap;
import java.util.Map;

import com.link.serve.tcp.message.Message;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.ProducerFactory;

import com.link.serve.kafka.proto.KafkaMessageProto;
import org.springframework.stereotype.Component;

@Component
public class DynamicKafkaProducerFactory {
    /**
     * IM下行类消息的生产者配置工厂
     * 构造一个 ProducerFactory，根据传入的 bootstrapServers、序列化器以及其他扩展配置来构造。
     * @param bootstrapServers Kafka 集群地址
     * @param keySerializer    Key 的序列化器，例如 org.apache.kafka.common.serialization.StringSerializer.class
     * @param valueSerializer  Value 的序列化器，例如 com.link.serve.kafka.coder.KafkaMessageSerializer.class
     * @param extraConfigs     其他扩展配置，可以为空
     * @return ProducerFactory 对象，泛型指定为 KafkaMessageProto.KafkaMessage
     */
    public static ProducerFactory<String, Message> createImMessageProducerFactory(
            String bootstrapServers,
            Class<?> keySerializer,
            Class<?> valueSerializer,
            Map<String, Object> extraConfigs) {

        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 生产者基础配置
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 50);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // 指定序列化器（动态配置）
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);

        // 如果有其他额外配置，则合并
        if (extraConfigs != null && !extraConfigs.isEmpty()) {
            props.putAll(extraConfigs);
        }

        // 返回类型指定为 KafkaMessageProto.KafkaMessage
        return new DefaultKafkaProducerFactory<>(props);
    }

    /**
     * Redis操作类消息的生产者配置工厂
     * 构造一个 ProducerFactory，根据传入的 bootstrapServers、序列化器以及其他扩展配置来构造。
     * @param bootstrapServers Kafka 集群地址
     * @param keySerializer    Key 的序列化器，例如 org.apache.kafka.common.serialization.StringSerializer.class
     * @param valueSerializer  Value 的序列化器，例如 com.link.serve.kafka.coder.KafkaMessageSerializer.class
     * @param extraConfigs     其他扩展配置，可以为空
     * @return ProducerFactory 对象，泛型指定为 KafkaMessageProto.KafkaMessage
     */
    public static ProducerFactory<String, KafkaMessageProto.KafkaMessage> createProducerFactory(
            String bootstrapServers,
            Class<?> keySerializer,
            Class<?> valueSerializer,
            Map<String, Object> extraConfigs) {

        Map<String, Object> props = new HashMap<>();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        // 生产者基础配置
        props.put(ProducerConfig.ACKS_CONFIG, "1");
        props.put(ProducerConfig.RETRIES_CONFIG, 3);
        props.put(ProducerConfig.BATCH_SIZE_CONFIG, 50);
        props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, 33554432);

        // 指定序列化器（动态配置）
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, keySerializer);
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, valueSerializer);

        // 如果有其他额外配置，则合并
        if (extraConfigs != null && !extraConfigs.isEmpty()) {
            props.putAll(extraConfigs);
        }

        // 返回类型指定为 KafkaMessageProto.KafkaMessage
        return new DefaultKafkaProducerFactory<>(props);
    }
}
